package cn.aohan.delayedqueue.provider;

import cn.aohan.delayedqueue.model.DelayedTaskInfo;
import cn.aohan.delayedqueue.model.TaskData;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;

/**
 * @author 傲寒
 * @date 2024/4/19
 */
@Component
public class DelayedQueueProvider {

    private final RedissonClient redissonClient;

    public DelayedQueueProvider(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    /**
     * 添加延迟任务
     *
     * @param val         值
     * @param delayTime   延迟时间
     * @param timeUnit    时间单位
     * @param delayedName 延迟名称
     */
    public void addDelayedTask(String delayedName, TaskData val, long delayTime, TimeUnit timeUnit) {
        final DelayedTaskInfo task = new DelayedTaskInfo();
        task.setCreateAt(System.currentTimeMillis());
        task.setDelayTime(delayTime);
        task.setTimeUnit(timeUnit);
        task.setVal(val);
        task.setDelayedName(delayedName);
        final RDelayedQueue<DelayedTaskInfo> delayedQueue = getDelayedQueue(delayedName);
        delayedQueue.offer(task, delayTime, timeUnit);
    }

    /**
     * 删除任务
     *
     * @param queueName 队列名称
     * @param taskId    任务id
     */
    public void removeTask(String queueName, String taskId) {
        final RBlockingDeque<DelayedTaskInfo> blockingDeque = getBlockingDeque(queueName);
        final Predicate<DelayedTaskInfo> predicate = item -> {
            final TaskData val = item.getVal();
            return Objects.nonNull(val) && Objects.equals(taskId, val.taskId);
        };
        blockingDeque.removeIf(predicate);
        final RDelayedQueue<DelayedTaskInfo> delayedQueue = getDelayedQueue(getBlockingDeque(queueName));
        delayedQueue.removeIf(predicate);
    }


    /**
     * 获取阻塞deque
     *
     * @param queueName 队列名称
     * @return {@link RBlockingDeque}<{@link DelayedTaskInfo}>
     */
    public RBlockingDeque<DelayedTaskInfo> getBlockingDeque(String queueName) {
        return redissonClient.getBlockingDeque(queueName, JsonJacksonCodec.INSTANCE);
    }

    /**
     * 获取延迟队列
     *
     * @param queueName 队列名称
     * @return {@link RDelayedQueue}<{@link DelayedTaskInfo}>
     */
    private RDelayedQueue<DelayedTaskInfo> getDelayedQueue(String queueName) {
        return redissonClient.getDelayedQueue(getBlockingDeque(queueName));
    }

    /**
     * 获取延迟队列
     *
     * @param blockingDeque 阻塞deque
     * @return {@link RDelayedQueue}<{@link DelayedTaskInfo}>
     */
    private RDelayedQueue<DelayedTaskInfo> getDelayedQueue(RBlockingDeque<DelayedTaskInfo> blockingDeque) {
        return redissonClient.getDelayedQueue(blockingDeque);
    }

}
